package com.whale.api;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.api.java.tuple.Tuple2;

/**
 * flink的流计算：数据是无界的
 *
 * 假如，我输入的顺序是： a b a b b
 * 观察到的
 * a(1)
 * b(1)
 * a(2)
 * b(2)
 * b(3)
 *
 * 它内部维护一个map，你来一次输入，它就整体记录一次，
 *
 * 而批处理不同，批处理的时候，数据是已知的，是能一口气全部能获取的，然后进行分组
 * 观察到：
 * a(2)
 * b(3)
 * 所以是一口气输出结果，而不是和流计算一样 一点一点
 *
 */
public class _01_StreamWordCount {

    public static void main(String[] args) throws Exception {

        // 流批一体的入口环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 本地运行模式时，程序的默认并行度为 ，你的cpu的逻辑核数
         */
        env.setParallelism(1);  // 默认并行度可以通过env人为指定

        // 通过source算子，把socket数据源加载为一个dataStream（数据流）
        SingleOutputStreamOperator<String> source = env.socketTextStream("localhost", 9999);

        // 然后通过算子对数据流进行各种转换（计算逻辑）
        SingleOutputStreamOperator<Tuple2<String, Integer>> words = source.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] arr = s.split("\\s+");

                for (String name : arr) {
                    collector.collect(Tuple2.of(name, 1));
                }
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> tuple2StringKeyedStream = words.keyBy(e -> e.f0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = tuple2StringKeyedStream.sum("f1");

        DataStreamSink<Tuple2<String, Integer>> print = sum.print();

        env.execute();
    }
}
